package org.niit.spark

import org.apache.spark.{SparkConf, SparkContext}

object HelloSpark {
  def main(args: Array[String]): Unit = {

    /*  spark  RDD    ; spark 分布式实时计算框架
       RDD  弹性分布式数据集  是 spark 数据类型 全称 （Resilient Distributed Dataset）弹性分布式数据集
       弹性   包含了计算逻辑，复杂，简单
       分布式   可以分布在不同的节点/机器,分布式存储
       数据集    比较类似 Array 或 List

      五大特性：
       1） 一个分区  partition
       2） 依赖列表   rdd ，计算 ，需要前后依赖     arrRdd.map().reduceByKey()
       3)  一个函数：  计算逻辑       arrRdd.map().reduceByKey()
       4) 可选的分区器
       5) 位置感知调度   对于 k-v 类型的  散列分区器，同时RDD 所在的每个分区都被记录了其所在的节点信息， 从而使的任务可以再就近的节点执行
     */

    // 创建    需要 jar包    1）通过 pom maven  下载依赖  2）直接下载好的jar包 导入
    //  创建 rdd     1） 从驱动器 Diver ( SparkContext ) 获取   2) 通过外部文件读取转化过来

    val conf = new SparkConf().setMaster("local[*]").setAppName("helloSpark")
    val sc = new SparkContext(conf)

    // array
    //    val array = Array(1,2,3,4,5 )
    //    val arrRdd =  sc.parallelize(array)
    //    println(arrRdd.sum())

    // spark 运行逻辑  Driver ->  节点 executor ，partition ，task   ---》 Driver

    // spark词频统计   hadoop 词频统计
    val arr = Array("hello spark","hello scala","hello java","hello hive")
    val arrRdd  = sc.parallelize(arr)
    // 1) 每个字段的数据切分，按照空格   ，每一个单词  hello，spark
    // 2) 映射   map  (hello ,1 )  (spark, 1)  (hello ,1 )  (scala, 1)
    // 3) 统计   key值 分组统计  (hello ,2)  ,(saprk  1 )

    //  最后通过 foreach循环打印， foreach( println(_) ) 打印相应的内容
    //arrRdd.flatMap(x => {x.split(" ")}).map(x => (x, 1)) .reduceByKey(_ + _).foreach(println(_))

    //      .reduceByKey(  (x:Int,y:Int) =>{ x+y}   )
    // spark  _ 通配符 ，表示集合里的每个元素
    //      .reduceByKey(_ +_   )

    //    map( x => (x,1))
    //    arrRdd.flatMap(x => { x.split(" ") }).map( (x:String)=> { return (x,1) }  )
    //    arrRdd.flatMap(x => { x.split(" ") }).map( (x:String)=> {  (x,1) }  )
    //    arrRdd.flatMap(x => { x.split(" ") }).map( (x:String)=>   (x,1)   )
    //    arrRdd.flatMap(x => { x.split(" ") }).map( x =>   (x,1)   )
    //    arrRdd.flatMap(x => { x.split(" ") }).map(  (_,1)   )

    //    arrRdd.flatMap(x => { x.split(" ") })

    //    arrRdd.flatMap((x:String) => { x.split(" ") })
    //    arrRdd.flatMap(x=> { x.split(" ") })
    //    arrRdd.flatMap(x=>  x.split(" ") )
    //    arrRdd.flatMap( _.split(" ") )
    // saprk 但凡写两行都不是好代码 ，一行
    println("------------")
    println("------------")
  val file=sc.textFile("D:\\DNIIT\\student.txt")
  val arrRdd2=sc.textFile("D:\\DNIIT\\student.txt")
  // file.flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).foreach(println(_))

    //spark  读取 student文件  统计top3的词频
    println("------1120-------")
    file.flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).foreach(println(_))
    file.flatMap(_.split(",")).map((_,1)).reduceByKey(_+_)
      .sortBy(_._2,false)
      //返回 集合 规定的 N 个元素
      .take(1).foreach(println(_))








































  }
}
